package com.atbeijing.bigdata.spark.core.rdd.operator.transform

import org.apache.spark.{SparkConf, SparkContext}

object Spark17_Oper_Transform {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("TransformOperator")
        val sc = new SparkContext(conf)

        // TODO 算子 - 转换 - KV - aggregateByKey (4 / 10)

        // aggregate表示聚合数据
        // UDAF
        // aggregateByKey算子用于相同key对value进行聚合
        // reduceByKey算子，分区内计算和分区间计算的逻辑是一样的。

        // TODO 数据集中相同key分区内取最大值，分区间求和
        // 【(a,1),(a,2),(b,3)】【(a,4),(b,5),(b,6)】
        // 【(a,2),(b,3)】【(a,4),(b,6)】
        // (a,6),(b,9)
        val rdd = sc.makeRDD(
            List(
                ("a", 1), ("a", 2), ("b", 3),
                ("a", 4), ("b", 5), ("b", 6)
            ),2
        )
        // aggregateByKey算子有函数柯里化操作，存在多个参数列表
        // 第一个参数列表有一个参数，这个参数表示计算初始值
        // 第二个参数列表有两个参数
        //     第一个参数表示 分区内相同key计算规则
        //     第二个参数表示 分区间相同key计算规则
        rdd.aggregateByKey(5)(
            (x, y) => math.max(x, y),
            (x, y) => x + y
        ).collect().foreach(println)
//        (b,11)
//        (a,10)
        println("================================")
        rdd.aggregateByKey(0)(
            (x, y) => x + y,
            (x, y) => x + y
        ).collect().foreach(println)
        println("================================")
        rdd.aggregateByKey(0)(_+_,_+_).collect().foreach(println)

        // TODO 如果aggregateByKey分区内计算逻辑和分区间计算逻辑相同的场合，可以采用另外一个算子代替
        // TODO foldByKey (5 / 10)
        rdd.foldByKey(0)(_+_).collect().foreach(println)

        // spark : reduceByKey(_+_), foldByKey(0)(_+_)
        // scala : reduce(_+_),      fold(0)(_+_)

        sc.stop()

    }
    class User {

    }
}
